package com.cloudcentral.websocket;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.TypeReference;
import com.cloudcentral.entity.User;
import com.cloudcentral.util.SpringContextUtils;
import com.cloudcentral.util.WebUtils;
import com.cloudcentral.websocket.entity.WSData;
import com.cloudcentral.websocket.entity.WSMessage;
import jakarta.websocket.*;
import jakarta.websocket.server.PathParam;
import jakarta.websocket.server.ServerEndpoint;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RLock;
import org.redisson.api.RSet;
import org.redisson.api.RedissonClient;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.TimeUnit;

import static com.cloudcentral.util.Utils.errorMessage;

/**
 * Create by Administrator
 * 2023/12/4 14:56
 */
@Slf4j
@Component
@EnableScheduling
@ServerEndpoint("/ws-cloud/{userId}")
public class CloudWebSocket {
  private static final List<Object> beans = new ArrayList<>();
  // concurrent包的线程安全Set,用来存放每个客户端对应的WebSocket对象。
  private static final CopyOnWriteArraySet<CloudWebSocket> webSocketSet = new CopyOnWriteArraySet<>();
  // session集合,存放对应的session
  private static final ConcurrentHashMap<Long, Session> sessionPool = new ConcurrentHashMap<>();
  private Long userId;
  // 与某个客户端的连接会话，需要通过它来给客户端发送数据
  private Session session;
  // 信号间隔最大时长
  public static final Duration duration = Duration.ofSeconds(15);
  // 上次通信时间戳
  public long lastTime;

  private static <T> T getBean(Class<T> clazz) {
    T bean = (T) beans.stream().filter(item -> item.getClass().equals(clazz)).findAny().orElse(null);
    if (bean == null) {
      bean = SpringContextUtils.getBean(clazz);
      beans.add(bean);
    }
    return bean;
  }

  private void done() throws Exception {
    session.close();
    sessionPool.remove(userId);
    webSocketSet.remove(this);
    RSet<Long> rSet = getBean(RedissonClient.class).getSet("WS-Cloud");
    rSet.remove(userId);
  }

  @Scheduled(initialDelay = 0, fixedRate = 1000)
  private void clearDone() {
    try {
      List<CloudWebSocket> list = webSocketSet.stream().filter(item -> System.currentTimeMillis() - item.lastTime > duration.toMillis()).toList();
      if (!list.isEmpty()) for (CloudWebSocket cloudWebSocket : list) cloudWebSocket.done();
    } catch (Exception e) {
      log.error("CloudWebSocket清理异常", e);
    }
  }

  @OnOpen
  public void onOpen(Session session, @PathParam("userId") Long userId) throws Exception {
    this.session = session;
    lastTime = System.currentTimeMillis();
    log.info("CloudWebSocket建立连接中,UserID：{}", userId);
    RLock lock = getBean(RedissonClient.class).getLock("WebSocket-Open");
    try {
      lock.tryLock(10, 10, TimeUnit.SECONDS);
      Session historySession = sessionPool.get(userId);
      // historySession不为空,说明已经有台站注册,应该删除登陆的WebSocket对象
      if (historySession != null) {
        webSocketSet.removeIf((item) -> item.session.equals(historySession));
        historySession.close();
      }
      // 建立连接
      this.userId = userId;
      webSocketSet.add(this);
      sessionPool.put(userId, session);
      log.info("CloudWebSocket建立连接成功,当前连接数为：{}", webSocketSet.size());
    } finally {
      lock.unlock();
    }
  }

  @OnError
  public void onError(Session session, Throwable e) {
    log.error("CloudWebSocket发生错误：{}", errorMessage(e));
    sendMessage(session, WSMessage.error(errorMessage(e)));
  }

  @OnClose
  public void onClose(Session session) throws Exception {
    done();
    log.info("CloudWebSocket连接断开,当前连接数为：{}", webSocketSet.size());
  }

  @OnMessage(maxMessageSize = 1024 * 1024 * 8)
  public void onMessage(Session session, String message) {
    lastTime = System.currentTimeMillis();
    RLock lock = getBean(RedissonClient.class).getLock("WebSocket-OnMessage");
    try {
      lock.tryLock(10, 10, TimeUnit.SECONDS);
      log.info("CloudWebSocket收到文本消息：{}", message);
      WSData<String> wsData = JSON.parseObject(message, new TypeReference<>() {
      });
      //心跳消息
      if (wsData.getType().equalsIgnoreCase(WSMessage.WSMessageType.HEART.getType())) sendMessage(session, WSMessage.heart());
    } catch (Exception e) {
      log.error("CloudWebSocket发送消息失败", e);
    } finally {
      lock.unlock();
    }
  }

  public static <T> void sendAllMessage(WSMessage<T> wsMessage) {
    for (CloudWebSocket webSocket : webSocketSet) sendMessage(webSocket.session, wsMessage);
  }

  public static <T> void sendMessageById(Long id, WSMessage<T> wsMessage) {
    sendMessage(sessionPool.get(id), wsMessage);
  }

  public static <T> void sendMessageForCurrentUser(WSMessage<T> wsMessage) {
    try {
      User user = WebUtils.currentUser();
      if (user.getEnable() && user.getId() != null) sendMessageById(user.getId(), wsMessage);
    } catch (Exception e) {
      log.error("CloudWebSocket 查找当前用户失败", e);
    }
  }

  public static <T> void sendMessage(Session session, WSMessage<T> wsMessage) {
    if (session != null) {
      String message = JSON.toJSONString(wsMessage);
      RLock lock = getBean(RedissonClient.class).getLock("WebSocket-SendMessage");
      try {
        lock.tryLock(10, 10, TimeUnit.SECONDS);
        //message = RSAUtil.privateKey().encrypt(message);
        session.getBasicRemote().sendText(message);
        log.info("CloudWebSocket发送消息: {}", message);
      } catch (Exception e) {
        log.error("CloudWebSocket发送消息失败", e);
      } finally {
        lock.unlock();
      }
    }
  }
}
